//
//  client.h
//  YCSB-C
//
//  Created by Jinglei Ren on 12/10/14.
//  Copyright (c) 2014 Jinglei Ren <jinglei@ren.systems>.
//

#ifndef YCSB_C_CLIENT_H_
#define YCSB_C_CLIENT_H_

#include <string>
#include "db.h"
#include "core_workload.h"
#include "utils.h"

namespace ycsbc {

class Client {
 public:
  Client(DB &db, CoreWorkload &wl) : db_(db), workload_(wl) { }
  
  virtual bool DoInsert(bool for_dhms = false, std::atomic<int>* insert_cnter = nullptr);
  virtual bool DoTransaction(bool for_dhms = false, std::atomic<int>* insert_cnter = nullptr);
  
  virtual ~Client() { }
  
 protected:
  
  virtual int TransactionRead(bool use_dhms = false, std::atomic<int>* cnter = nullptr);
  virtual int TransactionReadModifyWrite(bool use_dhms = false, std::atomic<int>* cnter = nullptr);
  virtual int TransactionScan();
  virtual int TransactionUpdate(bool use_dhms = false, std::atomic<int>* cnter = nullptr);
  virtual int TransactionInsert();
  
  DB &db_;
  CoreWorkload &workload_;
  int dhms_load_idx_ = 0;
};

// prepare data
inline bool Client::DoInsert(bool for_dhms, std::atomic<int>* insert_cnter) {
  if (!for_dhms) {
    std::string key = workload_.NextSequenceKey();
    std::vector<DB::KVPair> pairs;
    workload_.BuildValues(pairs);
    return (db_.Insert(workload_.NextTable(), key, pairs) == DB::kOK);
  } else {
    std::string oid = std::to_string(dhms_load_idx_);
    // field_len_generator_->Next(), utils::RandomPrintChar()
    std::string key(workload_.get_field_len_gen()->Next(), utils::RandomPrintChar());
    std::vector<DB::KVPair> pairs;
    int rc = db_.InsertAsync(oid, key, pairs, insert_cnter);
    dhms_load_idx_++;
    return (rc == DB::kOK);
  }
}

inline bool Client::DoTransaction(bool for_dhms, std::atomic<int>* cnter) {
  int status = -1;
  switch (workload_.NextOperation()) {
    case READ:
      status = TransactionRead(for_dhms, cnter);
      break;
    case UPDATE:
      status = TransactionUpdate(for_dhms, cnter);
      break;
    case INSERT:
      status = TransactionInsert();
      break;
    case SCAN:
      status = TransactionScan();
      break;
    case READMODIFYWRITE:
      status = TransactionReadModifyWrite(for_dhms, cnter);
      break;
    default:
      throw utils::Exception("Operation request is not recognized!");
  }
  assert(status >= 0);
  return (status == DB::kOK);
}

inline int Client::TransactionRead(bool use_dhms, std::atomic<int>* cnter) {
  if (use_dhms) {
    const std::string& oid = workload_.NextOid();
    std::string new_data;
    std::vector<DB::KVPair> result;
    return db_.ReadAsync(oid, new_data, NULL, result, cnter);
  } else {
    const std::string &table = workload_.NextTable();
    const std::string &key = workload_.NextTransactionKey();
    std::vector<DB::KVPair> result;
    if (!workload_.read_all_fields()) {
      std::vector<std::string> fields;
      fields.push_back("field" + workload_.NextFieldName());
      return db_.Read(table, key, &fields, result);
    } else {
      return db_.Read(table, key, NULL, result);
    } 
  }
}

inline int Client::TransactionReadModifyWrite(bool use_dhms, std::atomic<int>* cnter) {
  if (use_dhms) {
    const std::string& oid = workload_.NextOid();
    std::string key(workload_.get_field_len_gen()->Next(), utils::RandomPrintChar());
    std::vector<DB::KVPair> pairs;
    return db_.RMWAsync(oid, key, pairs, cnter);
  } else {
    const std::string &table = workload_.NextTable();
    const std::string &key = workload_.NextTransactionKey();
    std::vector<DB::KVPair> result;

    if (!workload_.read_all_fields()) {
      std::vector<std::string> fields;
      fields.push_back("field" + workload_.NextFieldName());
      db_.Read(table, key, &fields, result);
    } else {
      db_.Read(table, key, NULL, result);
    }

    std::vector<DB::KVPair> values;
    if (workload_.write_all_fields()) {
      workload_.BuildValues(values);
    } else {
      workload_.BuildUpdate(values);
    }
    return db_.Update(table, key, values);
  }
}

inline int Client::TransactionScan() {
  const std::string &table = workload_.NextTable();
  const std::string &key = workload_.NextTransactionKey();
  int len = workload_.NextScanLength();
  std::vector<std::vector<DB::KVPair>> result;
  if (!workload_.read_all_fields()) {
    std::vector<std::string> fields;
    fields.push_back("field" + workload_.NextFieldName());
    return db_.Scan(table, key, len, &fields, result);
  } else {
    return db_.Scan(table, key, len, NULL, result);
  }
}

inline int Client::TransactionUpdate(bool use_dhms, std::atomic<int>* cnter) {
  if (use_dhms) {
    const std::string& oid = workload_.NextOid();
    std::string key(workload_.get_field_len_gen()->Next(), utils::RandomPrintChar());
    std::vector<DB::KVPair> pairs;
    return db_.UpdateAsync(oid, key, pairs, cnter);
  } else {
    const std::string &table = workload_.NextTable();
    const std::string &key = workload_.NextTransactionKey();
    std::vector<DB::KVPair> values;
    if (workload_.write_all_fields()) {
      workload_.BuildValues(values);
    } else {
      workload_.BuildUpdate(values);
    }
    return db_.Update(table, key, values);
  }
}

inline int Client::TransactionInsert() {
  const std::string &table = workload_.NextTable();
  const std::string &key = workload_.NextSequenceKey();
  std::vector<DB::KVPair> values;
  workload_.BuildValues(values);
  return db_.Insert(table, key, values);
} 

} // ycsbc

#endif // YCSB_C_CLIENT_H_
